package org.dtp.alarm;

import org.dtp.DynamicThreadPoolExecutor;
import org.dtp.DynamicThreadPoolManager;
import org.dtp.config.DynamicThreadPoolProperties;
import org.dtp.config.ThreadPoolProperties;
import org.dtp.enums.AlarmEnum;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName DynamicThreadPoolAlarm
 * @Description 线程池告警
 * @Author wangyiqi
 * @Date 2021-08-11-20:38
 **/
public class DynamicThreadPoolAlarm {

    @Resource
    private DynamicThreadPoolManager dynamicThreadPoolManager;

    @Resource
    private DynamicThreadPoolProperties dynamicThreadPoolProperties;

    @Autowired(required = false)
    private ThreadPoolAlarmNotify threadPoolAlarmNotify;

    @PostConstruct
    public void init() {

        ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1, r -> {
            Thread thread = new Thread(r, "dynamic-thread-alarm");
            if (thread.isDaemon()) {
                thread.setDaemon(true);
            }
            if (thread.getPriority() != Thread.NORM_PRIORITY) {
                thread.setPriority(Thread.NORM_PRIORITY);
            }
            return thread;
        });
        scheduledExecutorService.scheduleWithFixedDelay(() -> dynamicThreadPoolProperties.getExecutors().forEach(prop -> {
            String threadPoolName = prop.getThreadPoolName();
            DynamicThreadPoolExecutor threadPoolExecutor = dynamicThreadPoolManager.getThreadPoolExecutor(threadPoolName);

            dynamicThreadPoolManager.registerStatusExtension(prop, threadPoolExecutor);

            alarmNotify(prop);

        }), 0, dynamicThreadPoolProperties.getRefreshTimeInterval(), TimeUnit.MILLISECONDS);
    }

    /**
     * 告警通知
     * @param threadPoolProperties /
     */
    private void alarmNotify(ThreadPoolProperties threadPoolProperties) {
        if (Objects.isNull(threadPoolAlarmNotify)) {
            return;
        }
        String threadPoolName = threadPoolProperties.getThreadPoolName();
        DynamicThreadPoolExecutor threadPoolExecutor = dynamicThreadPoolManager.getThreadPoolExecutor(threadPoolName);
        Integer rejectCount = Optional.ofNullable(dynamicThreadPoolManager.getRejectCount(threadPoolName)).map(AtomicInteger::get).orElse(0);
        int taskCount = threadPoolExecutor.getQueue().size();
        ArrayList<String> alarmReasons = new ArrayList<>();
        if (taskCount > threadPoolProperties.getQueueCapacityThreshold()) {
            alarmReasons.add(AlarmEnum.CAPACITY_ALARM.type());
        }
        if (rejectCount > 0) {
            alarmReasons.add(AlarmEnum.REJECT_ALARM.type());
        }
        if (CollectionUtils.isEmpty(alarmReasons)) {
            return;
        }
        AlarmMessage alarmMessage = new AlarmMessage.Builder()
                .alarmThreadName(threadPoolName)
                .alarmReasons(alarmReasons)
                .queueTaskCount(taskCount)
                .rejectTaskCount(rejectCount)
                .threadPoolProperties(threadPoolProperties)
                .build();
        threadPoolAlarmNotify.alarmNotify(alarmMessage);
    }

}
